package com.xxljob.demo.core.server;

import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.impl.ExecutorBizImpl;
import com.xxl.job.core.biz.model.*;
import com.xxl.job.core.thread.ExecutorRegistryThread;
import com.xxl.job.core.util.GsonTool;
import com.xxl.job.core.util.ThrowableUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.concurrent.*;

/**
 * @Author：YH
 * @Description：
 * @Date:Created in 2022/5/23 17:12
 */
@Service
public class EmbedServer {
    private static final Logger logger = LoggerFactory.getLogger(EmbedServer.class);

    private ExecutorBiz executorBiz;
    

    private Thread thread;

    public EmbedServer() {
    }

    public void start(final String address,
                      final int port,
                      final String appname,
                      final String accessToken) {
        this.executorBiz = new ExecutorBizImpl();
        this.thread = new Thread(new Runnable() {
            @Override
            public void run() {
                EventLoopGroup bossGroup = new NioEventLoopGroup();
                EventLoopGroup workerGroup = new NioEventLoopGroup();
                final ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(0, 200, 60L, TimeUnit.SECONDS,
                        new LinkedBlockingQueue(2000), new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "xxl-rpc, EmbedServer bizThreadPool-" + r.hashCode());
                    }
                }, new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");
                    }
                });

                try {
                    ServerBootstrap bootstrap = new ServerBootstrap();
                    bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline().addLast(new IdleStateHandler(0L, 0L, 90L,
                                    TimeUnit.SECONDS)).addLast(new HttpServerCodec()).addLast(new HttpObjectAggregator(5242880)).addLast(new EmbedHttpServerHandler(EmbedServer.this.executorBiz, accessToken, bizThreadPool));
                        }
                    }).childOption(ChannelOption.SO_KEEPALIVE, true);
                    //启动执行器端的监听端口
                    ChannelFuture future = bootstrap.bind(port).sync();
                    logger.info(">>>>>>>>>xxl-job remoting server start success,netttype={},port={}",
                            EmbedServer.class, port);
                    EmbedServer.this.startRegistry(appname, address);
                    future.channel().closeFuture().sync();
                } catch (Exception e) {
                    if (e instanceof InterruptedException) {
                        logger.info(">>>>>>>>>>> xxl-job remoting server stop.");
                    } else {
                        logger.error(">>>>>>>>>>> xxl-job remoting server error.", e);
                    }
                    e.printStackTrace();
                } finally {
                    try {
                        workerGroup.shutdownGracefully();
                        bossGroup.shutdownGracefully();
                    } catch (Exception var13) {
                        logger.error(var13.getMessage(), var13);
                    }
                }
            }
        });
        this.thread.setDaemon(true);
        this.thread.setName("EmbedServer Thread");
        this.thread.start();
    }

    public void startRegistry(String appname, String address) {
        ExecutorRegistryThread.getInstance().start(appname, address);
    }

    public static class EmbedHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
        private static final Logger logger =
                LoggerFactory.getLogger(com.xxl.job.core.server.EmbedServer.EmbedHttpServerHandler.class);
        private final ExecutorBiz executorBiz;
        private final String accessToken;
        private final ThreadPoolExecutor bizThreadPool;

        public EmbedHttpServerHandler(ExecutorBiz executorBiz, String accessToken, ThreadPoolExecutor bizThreadPool) {
            this.executorBiz = executorBiz;
            this.accessToken = accessToken;
            this.bizThreadPool = bizThreadPool;
        }

        @Override
        protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
            final String requestData = msg.content().toString(CharsetUtil.UTF_8);
            final String uri = msg.uri();
            final HttpMethod httpMethod = msg.method();
            final boolean keepAlive = HttpUtil.isKeepAlive(msg);
            final String accessTokenReq = msg.headers().get("XXL-JOB-ACCESS-TOKEN");
            logger.info("客户端接收到服务端调度的消息：{}", requestData);
            this.bizThreadPool.execute(new Runnable() {
                @Override
                public void run() {

                    Object responseObj = EmbedServer.EmbedHttpServerHandler.this.process(httpMethod, uri, requestData
                            , accessTokenReq);
                    String responseJson = GsonTool.toJson(responseObj);
                    logger.info("响应数据：{}", responseJson);
                    EmbedHttpServerHandler.this.writeResponse(ctx, keepAlive, responseJson);
                }
            });
        }

        private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) {
            if (HttpMethod.POST != httpMethod) {
                return new ReturnT(500, "invalid request, HttpMethod not support.");
            } else if (uri != null && uri.trim().length() != 0) {
                if (this.accessToken != null && this.accessToken.trim().length() > 0 && !this.accessToken.equals(accessTokenReq)) {
                    return new ReturnT(500, "The access token is wrong.");
                } else {
                    try {
                        if ("/beat".equals(uri)) {
                            ReturnT<String> beat = executorBiz.beat();
                            return this.executorBiz.beat();
                        } else if ("/idleBeat".equals(uri)) {
                            IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData,
                                    IdleBeatParam.class);
                            return this.executorBiz.idleBeat(idleBeatParam);
                        } else if ("/run".equals(uri)) {
                            TriggerParam triggerParam = GsonTool.fromJson(requestData,
                                    TriggerParam.class);
                            return this.executorBiz.run(triggerParam);
                        } else if ("/kill".equals(uri)) {
                            KillParam killParam = GsonTool.fromJson(requestData, KillParam.class);
                            return this.executorBiz.kill(killParam);
                        } else if ("/log".equals(uri)) {
                            LogParam logParam = GsonTool.fromJson(requestData, LogParam.class);
                            return this.executorBiz.log(logParam);
                        } else {
                            return new ReturnT(500, "invalid request, uri-mapping(" + uri + ") not found.");
                        }
                    } catch (Exception var6) {
                        logger.error(var6.getMessage(), var6);
                        return new ReturnT(500, "request error:" + ThrowableUtil.toString(var6));
                    }
                }
            } else {
                return new ReturnT(500, "invalid request, uri-mapping empty.");
            }
        }

        private void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson) {
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
                    Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8));
            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=UTF-8");
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
            if (keepAlive) {
                response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            }

            ctx.writeAndFlush(response);
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            logger.error(">>>>>>>>>>> xxl-job provider netty_http server caught exception", cause);
            ctx.close();
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                ctx.channel().close();
                logger.debug(">>>>>>>>>>> xxl-job provider netty_http server close an idle channel.");
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }
}
